package com.catmiao.sql;


import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.types.StringType;
import org.apache.spark.sql.types.StringType$;

public class Spark08_udf {

    public static void main(String[] args) {

        // 构建SparkSQL的环境对象时，一般采用构建器模式
        // 构造器模式：构建对象
        SparkSession sparkSession = SparkSession
                .builder()
                .master("local[*]")
                .appName("SparkSQL")
                .getOrCreate();


        // TODO Spark SQL中对数据模型也进行封装：RDD -> Dataset
        //      对接文件数据源时，会将文件中的一行数据封装为 Row对象
        Dataset<Row> ds = sparkSession.read().json("datas/sql/user.json");

        ds.createOrReplaceTempView("user");

        sparkSession.sql("select concat('Name:',username) from user").show();



        // TODO SparkSQL提供了一种特殊的方式，可以在SQL中增加自定义方法来实现复杂的逻辑


        // TODO 需要自定义方法在SQL中使用，需要在Spark中进行声明和注册
        //      register param1： SQL中使用的方法名
        //               param2， 逻辑
        //               param3  返回的数据类型：DataType类型数据，需要使用scala语法操作，需要特殊的使用方式

        sparkSession.udf().register("prefixName", new UDF1<String,String>() {
            @Override
            public String call(String s) throws Exception {
                return "UDF_NAME:" + s ;
            }
        }, StringType$.MODULE$);

        sparkSession.sql("select prefixName(username) from user").show();


        sparkSession.close();
    }

}
